package com.shujia.road

import java.lang
import java.time.Duration
import java.util.Properties

import com.shujia.bean.Car
import com.shujia.tf.CarMapper
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.state.StateBackend
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector
import redis.clients.jedis.Jedis

object RealTimeRoadAvgSpeed {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 每 1000ms 开始一次 checkpoint
    env.enableCheckpointing(10000)

    // 高级选项：

    // 设置模式为精确一次 (这是默认值)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

    // 确认 checkpoints 之间的时间会进行 500 ms
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

    // Checkpoint 必须在一分钟内完成，否则就会被抛弃
    env.getCheckpointConfig.setCheckpointTimeout(60000)

    // 同一时间只允许一个 checkpoint 进行
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

    // 开启在 job 中止后仍然保留的 externalized checkpoints
    env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

    // 允许在有更近 savepoint 时回退到 checkpoint
    env.getCheckpointConfig.setPreferCheckpointForRecovery(true)

    /*    val stateBackend: StateBackend = new RocksDBStateBackend("hdfs://master:9000/car/checkpoint", true)
        env.setStateBackend(stateBackend)*/


    //读取卡口过车数据
    val properties = new Properties
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    properties.setProperty("group.id", "asdasdasd")

    //创建消费者

    val flinkKafkaConsumer = new FlinkKafkaConsumer[String](
      "cars",
      new SimpleStringSchema,
      properties)

    flinkKafkaConsumer.setStartFromEarliest()


    val carsDS: DataStream[String] = env.addSource(flinkKafkaConsumer)


    val carDS: DataStream[Car] = carsDS.map(new CarMapper)

    //指定时间字段和水位线
    val assDS: DataStream[Car] = carDS.assignTimestampsAndWatermarks(//指定数据最大的延迟时间
      new BoundedOutOfOrdernessTimestampExtractor[Car](Time.seconds(1)) {
        //返回事件时间字段
        override def extractTimestamp(element: Car): Long = {
          element.getTime * 1000
        }
      })


    /**
      *1.3 实时统计道路拥堵情况-  实时计算每个道路平均速度，计算最近5分钟每个一分钟计算一次
      *
      */
    val keyByDS: KeyedStream[Car, lang.Long] = assDS.keyBy(_.getRoad_id)

    val windowDS: WindowedStream[Car, lang.Long, TimeWindow] = keyByDS.timeWindow(Time.minutes(5), Time.minutes(1))

    val avgSpeedDs: DataStream[(Long, Long, Double)] = windowDS.process(new ProcessWindowFunction[Car, (Long, Long, Double), lang.Long, TimeWindow] {
      override def process(key: lang.Long, context: Context, elements: Iterable[Car], out: Collector[(Long, Long, Double)]): Unit = {
        //获取所有车辆的车速
        var sumSpeed = 0.0
        var count = 0.0
        elements.foreach(car => {
          val speed: lang.Double = car.getSpeed
          sumSpeed += speed
          count += 1
        })

        val avgSpeed: Double = sumSpeed / count

        //获取窗口结束时间
        val endTime: Long = context.window.getEnd

        val keyScala: Long = key.toLong

        out.collect((keyScala, endTime, avgSpeed))
      }
    })

    //将数据保存到redis
    avgSpeedDs.addSink(new RichSinkFunction[(Long, Long, Double)] {
      var jedis: Jedis = _

      override def open(parameters: Configuration): Unit = {
        //1、创建redsi链接
        jedis = new Jedis("master", 6379)
      }

      override def close(): Unit = {
        jedis.close()
      }

      override def invoke(kv: (Long, Long, Double), context: SinkFunction.Context[_]): Unit = {
        val road: Long = kv._1
        val endTime: Long = kv._2
        val avgSpeed: Double = kv._3

        val key = s"RealTimeRoadAvgSpeed:$road:$endTime"

        jedis.set(key, avgSpeed.toString)

        //设置过期时间
        jedis.expire(key, 60 * 60)
      }
    })

    env.execute()

  }

}
